package com.tang.split;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 分流：奇数偶数拆分成不同流
 *
 * @author tang
 * @since 2023/6/9 10:42
 */
public class SplitByFilterDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("192.168.70.141", 7777);

        /*
         * 使用filter来进行分流
         * 缺点：同一个数据要被处理两遍
         */
        SingleOutputStreamOperator<Integer> map = source.map(Integer::valueOf);
        SingleOutputStreamOperator<Integer> even = map.filter(value -> value % 2 == 0);
        SingleOutputStreamOperator<Integer> odd = map.filter(value -> value % 2 == 1);
        even.print("偶数流");
        odd.print("奇数流");
        
        env.execute();
    }

}
